package baiduschPro;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import baidusch.BaiduSchDown;
import baidusch.FormatBaiduSchUrl;
import hylanda.common.FileWr;


@Tags({"baidusearch download"})
@CapabilityDescription("baidu search ,input content {\"schword\":\"nifi\",\"maxpage\":\"3\"}")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class BaiduSchProcessor extends AbstractProcessor {
	
	public static final PropertyDescriptor MYINTERVAL = new PropertyDescriptor
            .Builder().name("MYINTERVAL")
            .displayName("down interval")
            .description("down time interval,second")
            .required(true)
            .defaultValue("3")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

	 public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
	            .name("gen success")
	            .description("success relationship")
	            .build();
	    
	    public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
	            .name("gen failure")
	            .description("failure relationship")
	            .build();

	    private List<PropertyDescriptor> descriptors;

	    private Set<Relationship> relationships;

	    @Override
	    protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(MYINTERVAL);
	        this.descriptors = Collections.unmodifiableList(descriptors);

	        final Set<Relationship> relationships = new HashSet<Relationship>();
	        relationships.add(MY_RELATIONSHIPOK);
	        relationships.add(MY_RELATIONSHIPNO);
	        this.relationships = Collections.unmodifiableSet(relationships);
	        
	        
	    }

	    @Override
	    public Set<Relationship> getRelationships() {
	        return this.relationships;
	    }

	    @Override
	    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	        return descriptors;
	    }

	    @OnScheduled
	    public void onScheduled(final ProcessContext context) {

	  //  	System.out.println("onScheduled is clled");
	    }
	    
	
	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		FlowFile flowFile = session.get();
        if ( flowFile == null ) {
        	
        	System.out.println("flowfile is null");
           return;
        }
        //获取配置属性值
        PropertyValue timeintervalp = context.getProperty("MYINTERVAL");
        String timeinterval = timeintervalp.getValue();
        if(timeinterval == null)
        	timeinterval = "3";
        if(timeinterval.isEmpty())
        	timeinterval = "3";
        int interval = Integer.parseInt(timeinterval);
        
		//读取FlowFile
        String content = "";
        InputStream in = session.read(flowFile);
        try {
			byte[] by = readStream(in);
			content = new String(by,"UTF-8");
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
        //解析参数,生成url
        List urls = FormatBaiduSchUrl.FormaUrlsFromContent(content);
        String keyword = FormatBaiduSchUrl.getKeyWord(content);
        String downtime = "";
        //下载
        for(int i = 0 ; i < urls.size();i++)
        {
        	String url = (String) urls.get(i);
        	String page = BaiduSchDown.downOnePage(url);
       ///
 //     FileWr.saveFile("e:\\code\\baiduschtestnew.html", page);
       ///
        	downtime = getCurrentTime();
        	final String forFlowFile = BaiduSchDown.FormJson(keyword,url,page,downtime);
        	//写FlowFile
        	if(flowFile == null)
        		flowFile = session.create();
             
        	flowFile= session.write(flowFile, new OutputStreamCallback(){
        		@Override
        		public void process(OutputStream out) throws IOException{
        			out.write(forFlowFile.getBytes("UTF-8"));
        		}
        	});
            
            session.transfer(flowFile,MY_RELATIONSHIPOK);
            flowFile = null;
            
       //     session.co
            
            try {
				Thread.sleep(interval*1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        }
        
        
	}
	
	public static String getCurrentTime()
	{
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
		String str = df.format(new Date());// new Date()为获取当前系统时间
		return str;
	}
	 public static byte[] readStream(InputStream inStream) throws Exception {  
	        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
	        byte[] buffer = new byte[1024];  
	        int len = -1;  
	        while ((len = inStream.read(buffer)) != -1) {  
	            outSteam.write(buffer, 0, len);  
	        }  
	        outSteam.close();  
	        inStream.close();  
	        return outSteam.toByteArray();  
	    }  

}
